import dolphindb as ddb
import pandas as pd


class DdbData:
    """
    获取dolphindb数据
    """
    s = ddb.session()
    s.connect(host='101.95.132.98', port=33147, userid='CW', password='xqsy_cw')

    def __init__(self):
        pass

    # Fetch IV from DolphinDB
    @staticmethod
    def fetch_iv_from_dolphindb(s, sym: str = 'TA', month: str = '202311') -> pd.DataFrame:
        """
        写查询语句，完成dolphindb端数据的查询
        :param s: 数据库的链接
        :param sym: 合约代码：'TA','MA','RM','CF'...
        :param month: 合约月份：'202311','202312','202401'...
        :return: Dataframe
        """
        query = f'''
                iv = select *, abs(atm_loc) as abs_atm_loc, min(abs(atm_loc)) as s from 
                loadTable("dfs://impvResult", "settle_iv") where symbol = "{sym}" and month = "{month}" context by date;
                iv_atm = select * from iv where (abs_atm_loc-s) = 0;
                delete from iv_atm where iv = 0.35;
                select date, iv from iv_atm;
                '''
        return s.run(query)

    @staticmethod
    def iv_term_structure_from_dolphindb(s, sym: str = 'TA') -> pd.DataFrame:
        """
        查询获取波动率期限结构数据
        :param s: 数据库链接
        :param sym: 合约代码
        :return: Dataframe
        """
        query = f'''
                iv = select *, abs(atm_loc) as abs_atm_loc, min(abs(atm_loc)) as s from 
                loadTable("dfs://impvResult", "settle_iv") 
                where symbol = "{sym}" context by date,month;
                iv_atm = select * from iv where (abs_atm_loc-s) = 0;
                delete from iv_atm where iv = 0.35;
                select date,month,iv from iv_atm;
                '''
        return s.run(query)

    @staticmethod
    def option_end_date(s, sym: str = 'TA') -> pd.DataFrame:
        """
        查询指定品种逐月期权到期日
        :param s: 数据库链接
        :param sym: 合约代码
        :return: Dataframe
        """
        query = f'''
                select date(datetime) as date,code,underlying,endday from loadTable("dfs://option_info",`data) 
                where date(datetime) = max(date(datetime)),underlying like '{sym}%' context by date(datetime),underlying 
                limit 1
                '''
        return s.run(query)

    @staticmethod
    def iv_term_for_choose_month(s, sym: str = 'TA', month: tuple = ('11', '01')) -> pd.DataFrame:
        """
        获取指定合约，指定月份的历史波动率以及波动率差
        :param s: 数据库链接
        :param sym: 指定品种
        :param month: 指定两个月份
        :return: Dataframe
        """
        query = f'''
                iv = select *, abs(atm_loc) as abs_atm_loc, min(abs(atm_loc)) as s,substr(month,4,5) as m from 
                loadTable("dfs://impvResult", "settle_iv") where symbol = "{sym}" context by date,month;
                iv_atm = select * from iv where (abs_atm_loc-s) = 0;
                delete from iv_atm where iv = 0.35;
                data = lj((select date,iv,m,symbol from iv_atm where m='{month[0]}'),(select date,iv,m,symbol 
                from iv_atm where m='{month[1]}'),`symbol`date)
                select *,(iv-tmp2_iv) as term from data
                '''
        return s.run(query)

    @staticmethod
    def iv_data_all(s) -> pd.DataFrame:
        """
        通用数据获取：
        :return:
        """
        query = f'''
                iv = select *,abs(atm_loc) as abs_atm_loc,min(abs(atm_loc)) as s from 
                loadTable("dfs://impvResult",`settle_iv) context by date,symbol,month
                iv_atm = select * from iv where (abs_atm_loc-s) =0
                delete from iv_atm where iv =0.35
                select *,(iv-prev(iv)) as vi_R from iv_atm context by symbol,month
                '''
        return s.run(query)

    @staticmethod
    def iv_skew_all(s, sym: str = 'TA', month_new: str = '11') -> pd.DataFrame:
        """
        日度偏度数据获取：
        :return:
        """
        query = f'''
                data = select * from loadTable("dfs://factors",`iv_skew_date)  context by date,symbol,month
                select *,substr(month, 4, 2) as month_new from data where symbol =  "{sym}" , month_new =  "{month_new}" 
                context by date,symbol,month_new
                '''
        return s.run(query)

    @staticmethod
    def vol_int_future(s, sym: str = 'TA', month: str = '01') -> pd.DataFrame:
        """
        日度期货各合约的成交量和持仓量
        :return:
        """
        query = f'''
                data = select *  from loadTable("dfs://factors",`volume_openinterest) where  symbol = "{sym}" context by 
                date, instrumentid
                data_1 = select *, substr(data.instrumentid , regexFind (data.instrumentid  , "[0-9]", 3), 2)  as month 
                from data where isNull(opyiontype)
                select * from data_1 where month = "{month}"
                '''
        return s.run(query)

    @staticmethod
    def top_vol_int_option(s, sort_type: str = 'diff_vol') -> pd.DataFrame:
        """
        获取最新交易日成交量或者持仓量变化最大的前四个期权合约。
        :param s: 链接数据库
        :param sort_type: 排序筛选方式，默认成交量排序，diff_vol;如果选择持仓量排序，则需要将参数改为 diff_int
        :return: dataframe
        """
        query = f'''
                data = select * from loadTable("dfs://factors",`volume_openinterest) where date=(select max(date) 
                from loadTable("dfs://factors",`volume_openinterest)), isNull(opyiontype) = 0 order by date desc
                data_1 = select *, substr(data.instrumentid, regexFind (data.instrumentid, "[0-9]", 3), 2) as month 
                from data
                select top 4 instrumentid,symbol from data_1 context by symbol csort {sort_type} desc
                '''
        return s.run(query)

    @staticmethod
    def option_data_volume_openinterest(s, sym: str = 'TA', month: str = '01', option_type: str = 'C') -> pd.DataFrame:
        """
        日度期权成交量和持仓量获取，通过选定参数，获取需要的数据
        :param s: 链接数据库
        :param sym: 选定品种
        :param month: 选定月份
        :param option_type: 选择期权类型
        :return: dataframe
        """
        query = f'''
                data = select * from loadTable("dfs://factors",`volume_openinterest) where symbol = "{sym}" context by 
                date, instrumentid,opyiontype
                data_1 = select *, substr(data.instrumentid, regexFind (data.instrumentid , "[0-9]", 3), 2)  as month 
                from data where isNull(opyiontype) = 0       
                select date, sum(volume ) as sum_volume, sum(openinterest) as sum_openinterest from data_1 where
                month = "{month}", opyiontype = "{option_type}" group by date
                '''
        return s.run(query)

    @staticmethod
    def put_call_ratio(s, sym: str = 'TA') -> pd.DataFrame:
        """
        获取指定品种的期权数据并计算成交量pcr和持仓量pcr
        :return:
        """
        query = f'''
                data = select *  from loadTable("dfs://factors",`volume_openinterest) where symbol = "{sym}" context by 
                date,instrumentid,opyiontype
                data_1 = select *, substr(data.instrumentid, regexFind (data.instrumentid, "[0-9]", 3), 2) as month 
                from data where isNull(opyiontype) = 0       
                data_call = select date,symbol,month,sum(volume) as sum_volume,sum(openinterest) as sum_openinterest,
                opyiontype from data_1 where opyiontype = "C" group by date,symbol,month,opyiontype
                data_put = select date,symbol,month,sum(volume ) as sum_volume , sum(openinterest) as sum_openinterest,
                opyiontype from data_1 where opyiontype = "P" group by date,symbol,month, opyiontype
                data_new = lj(data_call,data_put,`date)
                select *, round(data_put_sum_volume*1.0 /sum_volume,4) as vol_pcr, 
                round(data_put_sum_openinterest / sum_openinterest,4) as interest_pcr from data_new  
                '''
        return s.run(query)

    @staticmethod
    def minute_data_load(s, symbol: str= 'MA401') -> pd.DataFrame:
        """
        从dolphindb获取分钟级期货历史行情数据。
        :param s:
        :param symbol:
        :return:
        """
        query = f'''
                select trigger,code,etfprice from loadTable("dfs://impvResult",`result) where code='MA401' 
                context by trigger limit 1
                '''
        return s.run(query)

    @staticmethod
    def high_low_iv_night_from_dolphindb(s) -> pd.DataFrame:
        """
        写查询语句，完成dolphindb端数据的查询
        :param s: 数据库的链接
        :param code: 合约代码：'TA402','MA402'...
        :return: Dataframe
        """
        query = f'''
                    login(`CW,`xqsy_cw)
    
                    '''读取curve_result 表  '''
                    temp_curveResult  =  select date(trigger) as date, minute(time(trigger)) as time, code, double(Kprice )as kprice,  impv from loadTable("dfs://impvResult", `curveResult )
                    
                    check_opentime = select * from temp_curveResult where time <=09:10:00.000,code= "CF309" context by date ,code ,time  limit 1
                    '''读取  option_info 的data 表，获取到期日'''
                    temp_option_info = select distinct underlying as code, endday from loadTable("dfs://option_info", `data )
                    
                    "统一option_info和curveresult的code"
                    code_for_curve = select distinct code from temp_curveResult
                    code = select * from  lj(code_for_curve,temp_option_info,`code) context by code limit 1 
                    
                    '''合并 code 和 curve_result表 , 获取各合约到期日 '''
                    temp_endday = lj(code,temp_curveResult,`code)
                    
                    // "原始表中没有的数据，合并表中却有数据，问题待挖掘"
                    // T1 = select * from temp_endday where code='MA404',date=2024.01.12
                    // T2 = select * from temp_curveResult where code='MA404',date=2023.01.12
                    
                    data_from_ctp = select TradingDay as date, minute(UpdateTime) as time,InstrumentID as code,LastPrice from loadTable("dfs://ctp_market_data",`ctp_market_data) where InstrumentID in (select code from code) context by TradingDay,InstrumentID,minute(UpdateTime) limit 1
                    
                    '''从合并表，定位虚两档合约 '''
                    ctp_info_join =select date, time, code, kprice,impv, LastPrice,abs(LastPrice - kprice) as  price_diff ,1..size(code) as index,endday  from  lj(temp_endday,data_from_ctp,`date`time`code) context by date,time,code 
                    //check_ctp_info_join = select * from ctp_info_join  where code="TA409" ,date = 2023.10.26
                    //atm_index= select *, min(price_diff) as  loc_atm , (index-2) as otm_p,  (index+2 )as otm_c , (endday-date) as lastdays  from ctp_info_join where price_diff =  min(price_diff) context by date,time,code 
                    
                    atm_index= select *, min(price_diff) as  loc_atm , (index-2) as otm_p,  (index+2 )as otm_c , (endday-date) as lastdays  from ctp_info_join  context by date,time,code 
                    
                    otm = select * from atm_index where price_diff =   loc_atm context by date,time,code limit 1
                    //atm_num_check = select *  from atm_index where code = 'CF309'
                    //atm_num_check = select *  from atm_num where date=2023.09.26, time=22:32m, code = 'MA311'
                    
                    otm_c = select *,impv as iv_c from lj(ctp_info_join,otm,`date`time`code) where index =otm_c
                    
                    '''因为有些时间点的平值合约就是表中顺序列中的第一个执行价，所以得不到位置为-1的看跌虚值合约 '''
                    otm_p = select *, impv as iv_p from lj(ctp_info_join,otm,`date`time`code) where index =otm_p
                    
                    //check_otm = select * from otm  where code= "CF309" ,otm_p<0
                    //
                    //check_otm_c = select * from otm_c  where code= "CF309"
                    //
                    //check_otm_p = select * from otm_p  where code= "CF309"
                    
                    ''' 取均值 ''' 
                    otm_avg = select* , ( iv_c+iv_p) /2 as avg_iv from lj(otm_c,otm_p,`date`time`code) where avg_iv > 0 
                    //check_otm_avg = select * from otm_avg  where code="TA409" 
                    
                    ''' 分日盘和夜盘   '''
                    otm_avg_day =  select date, time, code,kprice,lastdays,avg_iv,otm_avg.code.substr(0, regexFind(otm_avg.code, "[0-9]",1)) as symbol from otm_avg where 09:00m <=time <14:55m  
                    otm_avg_night=  select date, time, code,kprice,lastdays,avg_iv,otm_avg.code.substr(0, regexFind(otm_avg.code, "[0-9]",1)) as symbol  from otm_avg where 21:00m<=time <22:55m 
                    
                    
                    //check_day = select * from otm_avg where code= "CF309"
                    '''找日内最高IV     '''
                    otm_avg_day_high =select *,max(avg_iv) as high_iv from otm_avg_day where time >= 09:10m context by date,code 
                    otm_avg_night_high =select *,max(avg_iv) as high_iv from otm_avg_night where time >= 21:10m context by date,code 
                    
                    '''日内最高点时间  '''
                    day_high_time = select *,time as high_time,kprice as high_kprice from otm_avg_day_high where avg_iv = high_iv context by date,code csort time  limit 1
                    night_high_time = select *,time as high_time,kprice as high_kprice  from otm_avg_night_high where avg_iv = high_iv context by date,code csort time  limit 1
                       
                    '''开盘IV  '''
                    day_open= select *,min(avg_iv)  as open_iv from otm_avg_day where 09:05m<=time < 09:10m context by date,code 
                    night_open = select *,min(avg_iv) as open_iv from otm_avg_night where 21:05m<= time <21:10m context by date,code 
                    
                    '''开盘和低波、高波行数不一致，是因为数据不全  '''
                    day_open_time =  select *, time as open_time from day_open where avg_iv= open_iv  context by date,code  csort time  limit 1
                    night_open_time = select *, time as open_time from night_open where avg_iv= open_iv  context by date,code  csort time  limit 1
                    
                    //check_high = lj(otm_avg_day,day_high_time,`date`code)
                    
                    '''日内最高点后的最低IV , 低波和高波行数不一致，是因为数据不全 '''
                    otm_low_day =select *,min(avg_iv) as low_iv from  lj(otm_avg_day,day_high_time,`date`code) where time>high_time context by date,code 
                    otm_low_night =select *,min(avg_iv) as low_iv from  lj(otm_avg_night_high,night_high_time,`date`code) where time>high_time context by date,code
                    
                    day_low = select *,time as low_time from otm_low_day where avg_iv = low_iv context by date,code   csort time  limit 1
                    night_low = select * ,time as low_time from otm_low_night where avg_iv = low_iv context by date,code csort time  limit 1
                    
                    '''最高IV - 开盘IV = ins_iv , 最高IV - 升波后的最低IV = des_iv '''
                    
                    day_iv_diff = select  date, code,lastdays,symbol,high_time, low_time, open_iv,high_iv,low_iv,(high_iv-open_iv) as ins_iv, (high_iv - low_iv)as des_iv from lj(day_low,day_open_time,`date`code) where  ins_iv >=0.008, 0.04>des_iv>0,lastdays >=15 context by date,code limit 1
                    
                    night_iv_diff = select  date, code,lastdays,symbol,high_time,low_time, open_iv,high_iv,low_iv,(high_iv-open_iv) as ins_iv, (high_iv - low_iv)as des_iv from lj(night_low ,night_open_time,`date`code) where ins_iv >=0.008,0.04>des_iv>0 ,lastdays >=15 context by date,code limit 1
                    
                    
                    //day_iv_diff = select  date, code,lastdays,symbol,high_time, low_time, open_iv,high_iv,low_iv,(high_iv-open_iv) as ins_iv, (high_iv - low_iv)as des_iv from lj(day_low,day_open_time,`date`code) context by date,code limit 1
                    //
                    //night_iv_diff = select  date, code,lastdays,symbol,high_time,low_time, open_iv,high_iv,low_iv,(high_iv-open_iv) as ins_iv, (high_iv - low_iv)as des_iv from lj(night_low ,night_open_time,`date`code) context by date,code limit 1
                    
                    //check_day_time = select * from otm_avg_day  where  code="CF405" ,date = 2023.07.20
                    //check_day_time = select * from otm_avg_day  where  code="TA409" ,time < 09:10m context by date,time limit 1
                    //check_high_time = select * from day_high_time where code="CF405" 
                    //check_low_time = select * from day_low where code="CF405"
                    //check_open_time = select * from day_open_time where code="CF405"
                    '''垂直合并表   ''' 
                    ins_des_iv = unionAll(day_iv_diff,night_iv_diff)
                    
                    '''共享表   '''
                    share ins_des_iv as ins_des_iv_all
    
    
                '''
        return s.run(query)


    @staticmethod
    def ins_iv_day_from_dolphindb(s) -> pd.DataFrame:
        """
        写查询语句，完成dolphindb端数据的查询
        :param s: 数据库的链接
        :param code: 合约代码：'TA402','MA402'...
        :return: Dataframe
        """
        query = f'''
                login(`CW,`xqsy_cw)

                '''调用DB 技术分析（Technical Analysis）指标库 ''' 
                use ta
                
                '''  从共享表中(日内双卖中升波降波情况表，iv_new DB 文件生成)，获取出现升波的date、code、ins_iv 和 des_iv 列  '''
                date_code = select   date,code,ins_iv,des_iv from ins_des_iv_all order by date,code
                
                '''基于date_code表，切割成月份和品种，用于之后合并配对 '''
                date_month_code = select * , ("202" + substr(date_code.code , regexFind (date_code .code  , "[0-9]", 2),3) ) as month,date_code.code.substr(0, regexFind(date_code.code, "[0-9]",1)) as symbol from date_code
                
                ''' 读取settle_iv 表，获得标的收盘价、iv  '''
                settle_iv_close =select  date,symbol,month,iv,underlyingprice as close_price,abs(atm_loc),min(abs(atm_loc))as atm_index from loadTable("dfs://impvResult", `settle_iv) context by  date,symbol,month
                
                '''获取平值合约的iv ''' 
                atm_iv_close = select * from settle_iv_close where abs_atm_loc = atm_index
                
                ''' 合并共享表和settle_iv 表, 以共享表为基础，去合并，结果是相同的升波日，对应日度数据 '''
                ins_atm_iv= lj(date_month_code,atm_iv_close,`symbol`month)
                
                '''去除默认值 '''
                delete from ins_atm_iv where iv = 0.35
                
                '''求5日iv 均值，收盘价均值 '''
                avg_iv_price_mavg = select date,code,symbol,atm_iv_close_date,iv,mavg(iv,5) as mavg_iv_5,close_price,mavg(close_price,5) as mavg_price_5, ins_iv,des_iv,max(atm_iv_close_date) as max_date from ins_atm_iv  where date>atm_iv_close_date context by date,code 
                
                '''算RSI  '''
                avg_iv_price_rsi = select *,rsi(mavg_iv_5, timePeriod=14) as RSI_iv,rsi(mavg_price_5,timePeriod=14) as RSI_close from avg_iv_price_mavg 
                
                '''升波日前5天的RSI ''' 
                avg_iv_price_rsi_5 = select * from avg_iv_price_rsi where mavg_price_5>0 context by date,code  csort atm_iv_close_date limit -5
                    
        
                    '''
        return s.run(query)


test = DdbData()

# test.vol_int_all(DdbData.s).to_csv('vol_int.csv')
# data = test.vol_int_option_all(DdbData.s, sym ='TA',month='12')
# print(data)
# data = test.iv_term_for_choose_month(DdbData.s,'TA','11')
# data = test.vol_int_future(DdbData.s,'TA')
# print(test.put_call_ratio(DdbData.s))
